/*
 * Copyright (C) 2018 Andreas Granig (sipwise.com)
 *
 * This file is part of Kamailio, a free SIP server.
 *
 * Kamailio is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version
 *
 * Kamailio is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 */

#include <stdlib.h>
#include <time.h>

#include "db_redis_mod.h"
#include "redis_connection.h"
#include "redis_dbase.h"
#include "redis_table.h"

static void db_redis_dump_reply(redisReply *reply) {
    int i;
    if (reply->type == REDIS_REPLY_STRING) {
        LM_DBG("%s\n", reply->str);
    } else if (reply->type == REDIS_REPLY_INTEGER) {
        LM_DBG("%lld\n", reply->integer);
    } else if (reply->type == REDIS_REPLY_NIL) {
        LM_DBG("<null>\n");
    } else if (reply->type == REDIS_REPLY_ARRAY) {
        LM_DBG("printing %lu elements in array reply\n", reply->elements);
        for(i = 0; i < reply->elements; ++i) {
            db_redis_dump_reply(reply->element[i]);
        }
    } else {
        LM_DBG("not printing invalid reply type\n");
    }
}

// TODO: utilize auto-expiry? on insert/update, also update expire value
// of mappings

/*
 * Initialize database module
 * No function should be called before this
 */
db1_con_t* db_redis_init(const str* _url) {
    return db_do_init(_url, (void *)db_redis_new_connection);
}

/*
 * Shut down database module
 * No function should be called after this
 */
void db_redis_close(db1_con_t* _h) {
    LM_DBG("closing redis db connection\n");
    db_do_close(_h, db_redis_free_connection);
}


static db1_res_t* db_redis_new_result(void) {
    db1_res_t* obj;

    obj = db_new_result();
    if (!obj)
        return NULL;
    return obj;
}

static int db_redis_val2str(const db_val_t *v, str *_str) {
    const char *s;
    const str *tmpstr;
    int vtype = VAL_TYPE(v);
    _str->s = NULL;
    _str->len = 32; // default for numbers

    if (VAL_NULL(v)) {
        LM_DBG("converting <null> value to str\n");
        _str->len = 0;
        return 0;
    }

    switch (vtype) {
        case DB1_INT:
            LM_DBG("converting int value %d to str\n", VAL_INT(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%d", VAL_INT(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_BIGINT:
            LM_DBG("converting bigint value %lld to str\n", VAL_BIGINT(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%lld", VAL_BIGINT(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_STRING:
            s = VAL_STRING(v);
            _str->len = strlen(s);
            LM_DBG("converting string value '%s' with len %d to str\n", s, _str->len);
            _str->s = (char*)pkg_malloc(_str->len + 1);
            if (!_str->s) goto memerr;
            //memcpy(_str->s, s, _str->len);
            //_str->s[_str->len] = '\0';
            memset(_str->s, 0, _str->len + 1);
            strncpy(_str->s, s, _str->len);
            break;
        case DB1_STR:
            tmpstr = &(VAL_STR(v));
            LM_DBG("converting str value '%.*s' with len %d to str\n", tmpstr->len, tmpstr->s, tmpstr->len);
            // copy manually to add 0 termination
            _str->s = (char*)pkg_malloc(tmpstr->len + 1);
            if (!_str->s) goto memerr;
            _str->len = tmpstr->len;
            memcpy(_str->s, tmpstr->s, _str->len);
            _str->s[_str->len] = '\0';
            break;
        case DB1_DATETIME:
            LM_DBG("converting datetime value %ld to str\n", VAL_TIME(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            strftime(_str->s, _str->len, "%Y-%m-%d %H:%M:%S", localtime(&(VAL_TIME(v))));
            _str->len = strlen(_str->s);
            break;
        case DB1_DOUBLE:
            LM_DBG("converting double value %f to str\n", VAL_DOUBLE(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%.6f", VAL_DOUBLE(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_BITMAP:
            LM_DBG("converting bitmap value %u to str\n", VAL_BITMAP(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%u", VAL_BITMAP(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_BLOB:
        default:
            LM_ERR("Unsupported val type %d\n", vtype);
            goto err;
    }

    return 0;

memerr:
    LM_ERR("Failed to allocate memory to convert value to string\n");
err:
    return -1;
}

static int db_redis_return_version(const db1_con_t* _h, km_redis_con_t *con, const str *table_name,
    db1_res_t **_r) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    db_val_t* dval;
    db_row_t* drow;

    LM_DBG("get table version\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define it in schema file!\n",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    *_r = db_redis_new_result();
    if (!*_r) {
        LM_ERR("Failed to allocate memory for result");
        return -1;
    }
    RES_NUM_ROWS(*_r) = 1;
    RES_COL_N(*_r) = 1;
    RES_ROW_N(*_r) = 1;
    if (db_allocate_rows(*_r) != 0) {
        LM_ERR("Failed to allocate memory for rows\n");
        goto err;
    }
    if (db_allocate_columns(*_r, 1) != 0) {
        LM_ERR("Failed to allocate memory for result columns");
        goto err;
    }

    drow = &(RES_ROWS(*_r)[0]);

    if (db_allocate_row(*_r, drow) != 0) {
        LM_ERR("Failed to allocate row %d\n", RES_NUM_ROWS(*_r));
        goto err;
    }

    dval = &(ROW_VALUES(drow)[0]);

    VAL_TYPE(dval) = DB1_INT;
    VAL_NULL(dval) = 0;
    VAL_INT(dval) = table->version;

    LM_DBG("returning short-cut table version %d for table '%.*s'",
        table->version, table_name->len, table_name->s);

    return 0;

err:
    if (*_r) db_redis_free_result((db1_con_t*)_h, *_r);
    return -1;
}

static int db_redis_build_entry_manual_keys(redis_table_t *table, const db_key_t *_k,
        const db_val_t *_v, const int _n, int **manual_keys, int *manual_key_count) {

    // TODO: we also put keys here which are already part of type mapping!
    // there must be removed for performance reasons

    redis_key_t *key = NULL;

    *manual_keys = (int*)pkg_malloc(_n * sizeof(int));
    if (! *manual_keys) {
        LM_ERR("Failed to allocate memory for manual key indices\n");
        goto err;
    }
    memset(*manual_keys, 0, _n * sizeof(int));
    *manual_key_count = 0;

    for (key = table->entry_keys; key; key = key->next) {
        int subkey_found = 0;
        int i;
        LM_DBG("checking for existence of entry key '%.*s' in query to get manual key\n",
                key->key.len, key->key.s);
        for (i = 0; i < _n; ++i) {
            const db_key_t k = _k[i];
            if (!str_strcmp(&key->key, (str*)k)) {
                LM_DBG("found key in entry key\n");
                subkey_found = 1;
                break;
            } else {
                (*manual_keys)[*manual_key_count] = i;
                (*manual_key_count)++;
            }
        }
        if (!subkey_found) {
            break;
        }
    }
    return 0;

err:
    if (*manual_keys) {
        pkg_free(*manual_keys);
        *manual_keys = NULL;
    }
    return -1;
}

static int db_redis_find_query_key(redis_key_t *key, const str *table_name,
        str *type_name, const db_key_t *_k, const db_val_t *_v, const int _n,
        str *key_name, int *key_found) {

    unsigned int len;
    str val = {NULL, 0};

    *key_found = 1;
    key_name->len = 0;
    key_name->s = NULL;

    for (; key; key = key->next) {
        int subkey_found = 0;
        int i;
        LM_DBG("checking for existence of entry key '%.*s' in query\n",
                key->key.len, key->key.s);
        for (i = 0; i < _n; ++i) {
            const db_key_t k = _k[i];
            const db_val_t v = _v[i];

            if (VAL_NULL(&v)) {
                LM_DBG("Skipping null value for given key '%.*s'\n",
                        k->len, k->s);
                break;
            } else if (!str_strcmp(&key->key, (str*)k)) {
                LM_DBG("found key in entry key\n");
                if (db_redis_val2str(&v, &val) != 0) goto err;
                if (val.s == NULL) {
                    LM_DBG("key value in entry key is null, skip key\n");
                    subkey_found = 0;
                    break;
                }
                if (!key_name->len) {
                    // <table_name>:<type>::<val>
                    len = table_name->len + 1 + type_name->len + 2 + val.len + 1; //snprintf writes term 0 char
                    key_name->s = (char*)pkg_malloc(len);
                    if (!key_name->s) {
                        LM_ERR("Failed to allocate key memory\n");
                        goto err;
                    }
                    snprintf(key_name->s, len, "%.*s:%.*s::%.*s",
                            table_name->len, table_name->s,
                            type_name->len, type_name->s,
                            val.len, val.s);
                    key_name->len = len-1; // subtract the term 0 char

                } else {
                    // :<val>
                    key_name->s = (char*)pkg_realloc(key_name->s, key_name->len + val.len + 2);
                    if (!key_name->s) {
                        LM_ERR("Failed to allocate key memory\n");
                        goto err;
                    }
                    snprintf(key_name->s + key_name->len, 1 + val.len + 1, ":%.*s",
                            val.len, val.s);
                    key_name->len += (1 + val.len);
                }
                LM_DBG("entry key so far is '%.*s'\n", key_name->len, key_name->s);
                subkey_found = 1;
                pkg_free(val.s);
                val.s = NULL;
                break;
            }
        }
        if (!subkey_found) {
            LM_DBG("key '%.*s' for type '%.*s' not found, unable to use this type\n",
                    key->key.len, key->key.s, type_name->len, type_name->s);
            if (key_name->s) {
                pkg_free(key_name->s);
                key_name->s = NULL;
                key_name->len = 0;
            }
            *key_found = 0;
            break;
        }
    }

    return 0;

err:
    if (val.s)
        pkg_free(val.s);
    if(key_name->s) {
        pkg_free(key_name->s);
        key_name->s = NULL;
        key_name->len = 0;
    }
    return -1;
}

static int db_redis_build_entry_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const db_val_t *_v, const int _n,
        redis_key_t **keys, int *keys_count) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_key_t *key;
    int key_found;
    str type_name = str_init("entry");
    str keyname = {NULL, 0};

    LM_DBG("build entry keys\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;
    key = table->entry_keys;
    if (db_redis_find_query_key(key, table_name, &type_name, _k, _v, _n, &keyname, &key_found) != 0) {
        goto err;
    }
    if (key_found) {
        if (db_redis_key_add_str(keys, &keyname) != 0) {
            LM_ERR("Failed to add key string\n");
            goto err;
        }
        LM_DBG("found suitable entry key '%.*s' for query\n",
                (*keys)->key.len, (*keys)->key.s);
        *keys_count = 1;
        pkg_free(keyname.s);
    } else {
        LM_ERR("Failed to create direct entry key, no matching key definition\n");
        goto err;
    }

    return 0;

err:
    db_redis_key_free(keys);
    if (keyname.s)
        pkg_free(keyname.s);
    return -1;
}

static int db_redis_get_keys_for_all_types(km_redis_con_t *con, const str *table_name,
        redis_key_t **keys, int *keys_count) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_type_t *type;
    redis_key_t *key;

    *keys = NULL;
    *keys_count = 0;

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    for (type = table->types; type; type = type->next) {
        for (key = type->keys; key; key = key->next) {
            if (db_redis_key_add_str(keys, &key->key) != 0) {
                LM_ERR("Failed to add key string\n");
                goto err;
            }
            (*keys_count)++;
        }
    }

    return 0;

err:
    db_redis_key_free(keys);
    return -1;
}

static int db_redis_build_type_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const db_val_t *_v, const int _n,
        redis_key_t **keys, int *keys_count) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_type_t *type;
    redis_key_t *key;

    *keys = NULL;
    *keys_count = 0;

    LM_DBG("build type keys\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    for (type = table->types; type; type = type->next) {
        str *type_name = &(type->type);
        int key_found = 0;
        str keyname = {NULL, 0};
        key = type->keys;

        if (db_redis_find_query_key(key, table_name, &type->type, _k, _v, _n, &keyname, &key_found) != 0) {
            goto err;
        }
        if (key_found) {
            if (db_redis_key_add_str(keys, &keyname) != 0) {
                LM_ERR("Failed to add query key to key list\n");
                goto err;
            }
            (*keys_count)++;
            LM_DBG("found key '%.*s' for type '%.*s'\n",
                    keyname.len, keyname.s,
                    type_name->len, type_name->s);
            pkg_free(keyname.s);
        }
    }

    return 0;

err:
    LM_ERR("Failed to get type key\n");
    db_redis_key_free(keys);
    return -1;
}

static int db_redis_build_query_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const db_val_t *_v, const db_op_t *_op, const int _n,
        redis_key_t **query_keys, int *query_keys_count, int **manual_keys, int *manual_keys_count,
        int *do_table_scan) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_type_t *type;
    redis_key_t *key;
    str keyname;
    int key_found;
    redisReply *reply = NULL;
    str typename = str_init(REDIS_DIRECT_PREFIX);

    *query_keys = NULL;
    *query_keys_count = 0;
    *do_table_scan = 1;

    LM_DBG("build query keys\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    // check if given keys directly match entry key
    keyname.s = NULL;
    keyname.len = 0;
    key = table->entry_keys;

    if (db_redis_find_query_key(key, table_name, &typename, _k, _v, _n, &keyname, &key_found) != 0) {
        goto err;
    }
    if (key_found) {
        LM_DBG("found suitable entry key '%.*s' for query\n",
                keyname.len, keyname.s);
        if (db_redis_key_add_str(query_keys, &keyname) != 0) {
            LM_ERR("Failed to add key name to query keys\n");
            goto err;
        }
        *query_keys_count = 1;
        pkg_free(keyname.s);
        keyname.s = NULL;
    } else {
        LM_DBG("no direct entry key found, checking type keys\n");
        for (type = table->types; type; type = type->next) {
            key = type->keys;
            LM_DBG("checking type '%.*s'\n", type->type.len, type->type.s);
            if (db_redis_find_query_key(key, table_name, &type->type, _k, _v, _n, &keyname, &key_found) != 0) {
                goto err;
            }
            if (key_found) {
                redis_key_t *query_v = NULL;
                char *prefix = "SMEMBERS";

                if (db_redis_key_add_string(&query_v, prefix, strlen(prefix)) != 0) {
                    LM_ERR("Failed to add smembers command to query\n");
                    db_redis_key_free(&query_v);
                    goto err;
                }
                if (db_redis_key_add_str(&query_v, &keyname) != 0) {
                    LM_ERR("Failed to add key name to smembers query\n");
                    db_redis_key_free(&query_v);
                    goto err;
                }

                reply = db_redis_command_argv(con, query_v);
                pkg_free(keyname.s);
                keyname.s = NULL;
                db_redis_key_free(&query_v);
                db_redis_check_reply(con, reply, err);
                if (reply->type == REDIS_REPLY_ARRAY) {
                    if (reply->elements == 0) {
                        LM_DBG("type query returned empty list\n");
                        *query_keys_count = 0;
                        *do_table_scan = 0;
                        db_redis_free_reply(&reply);
                        break;
                    } else {
                        int i;
                        LM_DBG("populating query keys list with result of type query\n");
                        *query_keys_count = reply->elements;
                        for (i = 0; i < reply->elements; ++i) {
                            redisReply *subreply = reply->element[i];
                            if (subreply->type == REDIS_REPLY_STRING) {
                                LM_DBG("adding resulting entry key '%s' from type query\n", subreply->str);
                                if (db_redis_key_add_string(query_keys, subreply->str, strlen(subreply->str)) != 0) {
                                    LM_ERR("Failed to add query key\n");
                                    goto err;
                                }
                            } else {
                                LM_ERR("Unexpected entry key type in type query, expecting a string\n");
                                goto err;
                            }
                        }
                    }
                } else {
                    LM_ERR("Unexpected reply for type query, expecting an array\n");
                    goto err;
                }

                db_redis_free_reply(&reply);
                break;
            }
        }
    }

    if (*query_keys_count > 0) {
        LM_DBG("building manual keys\n");
        if (db_redis_build_entry_manual_keys(table, _k, _v, _n, manual_keys, manual_keys_count) != 0) {
            LM_ERR("Failed to build manual entry key list\n");
            goto err;

        }
    }

    return 0;
err:
    if (keyname.s) {
        pkg_free(keyname.s);
        keyname.s = NULL;
    }
    if (reply) {
        db_redis_free_reply(&reply);
    }
    db_redis_key_free(query_keys);
    if (*manual_keys) {
        pkg_free(*manual_keys);
        *manual_keys = NULL;
    }
    return -1;
}

static int db_redis_scan_query_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const int _n,
        redis_key_t **query_keys, int *query_keys_count,
        int **manual_keys, int *manual_keys_count) {

    size_t i = 0;
    redis_key_t *query_v = NULL;
    char cursor_str[32] = "";
    redisReply *reply = NULL;
    unsigned long cursor = 0;
    char *match = NULL;
    size_t j;
    int l;

    str match_pattern = {":entry::*", strlen(":entry::*")};

    *query_keys = NULL;
    *query_keys_count = 0;
    *manual_keys = NULL;
    *manual_keys_count = 0;

    do {
        snprintf(cursor_str, sizeof(cursor_str), "%lu", cursor);
        match = (char*)pkg_malloc(table_name->len + match_pattern.len + 1);
        if (!match) {
            LM_ERR("Failed to allocate memory for match pattern\n");
            goto err;
        }
        snprintf(match, table_name->len + match_pattern.len + 1, "%s%s\n",
                table_name->s, match_pattern.s);

        if (db_redis_key_add_string(&query_v, "SCAN", 4) != 0) {
            LM_ERR("Failed to add scan command to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, cursor_str, strlen(cursor_str)) != 0) {
            LM_ERR("Failed to add cursor to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, "MATCH", 5) != 0) {
            LM_ERR("Failed to add match command to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, match, strlen(match)) != 0) {
            LM_ERR("Failed to add match pattern to scan query\n");
            goto err;
        }
        pkg_free(match); match = NULL;

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, err);
        if (reply->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Invalid reply type for scan on table '%.*s', expected array\n",
                    table_name->len, table_name->s);
            goto err;
        }
        if (reply->elements != 2) {
            LM_ERR("Invalid number of reply elements for scan on table '%.*s', expected 2, got %lu\n",
                    table_name->len, table_name->s, reply->elements);
            goto err;
        }

        if (reply->element[0]->type == REDIS_REPLY_STRING) {
            cursor = atol(reply->element[0]->str);
        } else if (reply->element[0]->type == REDIS_REPLY_INTEGER) {
            // should not happen, but play it safe
            cursor = reply->element[0]->integer;
        } else {
            LM_ERR("Invalid cursor type for scan on table '%.*s', expected string or integer\n",
                    table_name->len, table_name->s);
            goto err;
        }

        if (reply->element[1]->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Invalid content type for scan on table '%.*s', expected array\n",
                    table_name->len, table_name->s);
            goto err;
        }
        if (reply->element[1]->elements == 0) {
            LM_DBG("no matching entries found for scan on table '%.*s'\n",
                    table_name->len, table_name->s);
            return 0;
        }

        *query_keys_count += reply->element[1]->elements;

        for (j = 0; j < reply->element[1]->elements; ++i, ++j) {
            redisReply *key = reply->element[1]->element[j];
            if (!key) {
                LM_ERR("Invalid null key at cursor result index %lu while scanning table '%.*s'\n",
                        j, table_name->len, table_name->s);
                goto err;
            }
            if (key->type != REDIS_REPLY_STRING) {
                LM_ERR("Invalid key type at cursor result index %lu while scanning table '%.*s', expected string\n",
                        j, table_name->len, table_name->s);
                goto err;
            }
            if (db_redis_key_add_string(query_keys, key->str, strlen(key->str)) != 0) {
                LM_ERR("Failed to add redis key\n");
                goto err;
            }
        }
        db_redis_free_reply(&reply);
    } while (cursor > 0);

    // for full table scans, we have to manually match all given keys
    *manual_keys_count = _n;
    *manual_keys = (int*)pkg_malloc(*manual_keys_count * sizeof(int));
    if (! *manual_keys) {
        LM_ERR("Failed to allocate memory for manual keys\n");
        goto err;
    }
    memset(*manual_keys, 0, *manual_keys_count * sizeof(int));
    for (l = 0; l < _n; ++l) {
        (*manual_keys)[l] = l;
    }

    if (reply) {
        db_redis_free_reply(&reply);
    }
    return 0;

err:
    if (match)
        pkg_free(match);
    if (reply)
        db_redis_free_reply(&reply);
    db_redis_key_free(&query_v);
    db_redis_key_free(query_keys);
    *query_keys_count = 0;
    if (*manual_keys) {
        pkg_free(*manual_keys);
        *manual_keys = NULL;
    }
    return -1;
}

static int db_redis_compare_column(db_key_t k, db_val_t *v, db_op_t op, redisReply *reply) {
    int i_value;
    long long ll_value;
    double d_value;
    str *tmpstr;
    char tmp[32] = "";

    int vtype = VAL_TYPE(v);

    if (VAL_NULL(v) && reply->type == REDIS_REPLY_NIL) {
        LM_DBG("comparing matching NULL values\n");
        return 0;
    } else if (VAL_NULL(v) || reply->type == REDIS_REPLY_NIL) {
        LM_DBG("comparing non-matching NULL values\n");
        return -1;
    }

    switch (vtype) {
        case DB1_INT:
            i_value = atoi(reply->str);
            LM_DBG("comparing INT %d %s %d\n", i_value, op, VAL_INT(v));
            if (!strcmp(op, OP_EQ)) {
                if (i_value == VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_LT)) {
                if (i_value < VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_GT)) {
                if (i_value > VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_LEQ)) {
                if (i_value <= VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_GEQ)) {
                if (i_value >= VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_NEQ)) {
                if (i_value != VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_BITWISE_AND)) {
                if (i_value & VAL_INT(v))
                    return 0;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_BIGINT:
            ll_value = atoll(reply->str);
            LM_DBG("comparing BIGINT %lld %s %lld\n", ll_value, op, VAL_BIGINT(v));
            if (!strcmp(op, OP_EQ)) {
                if (ll_value == VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_LT)) {
                if (ll_value < VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_GT)) {
                if (ll_value > VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_LEQ)) {
                if (ll_value <= VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_GEQ)) {
                if (ll_value >= VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_NEQ)) {
                if (ll_value != VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_BITWISE_AND)) {
                if (ll_value & VAL_BIGINT(v))
                    return 0;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_STRING:
            LM_DBG("comparing STRING %s %s %s\n", reply->str, op, VAL_STRING(v));
            if (!strcmp(op, OP_EQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) == 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (strcmp(reply->str, VAL_STRING(v)) < 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (strcmp(reply->str, VAL_STRING(v)) > 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) <= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) >= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) != 0) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_STR:
            tmpstr = (struct _str*) &(VAL_STR(v));
            LM_DBG("comparing STR %s %s %.*s\n", reply->str, op, tmpstr->len, tmpstr->s);
            if (!strcmp(op, OP_EQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) == 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) < 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) > 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) <= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) >= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) != 0) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_DOUBLE:
            d_value = atof(reply->str);
            LM_DBG("comparing DOUBLE %f %s %f\n", d_value, op, VAL_DOUBLE(v));
            if (!strcmp(op, OP_EQ)) {
                return (d_value == VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (d_value < VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (d_value > VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (d_value <= VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (d_value >= VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (d_value != VAL_DOUBLE(v)) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_DATETIME:
            // TODO: insert int value to db for faster comparison!
            strftime(tmp, sizeof(tmp), "%Y-%m-%d %H:%M:%S", localtime(&(VAL_TIME(v))));
            LM_DBG("comparing DATETIME %s %s %s\n", reply->str, op, tmp);
            if (!strcmp(op, OP_EQ)) {
                return (strcmp(reply->str, tmp) == 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (strcmp(reply->str, tmp) < 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (strcmp(reply->str, tmp) > 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (strcmp(reply->str, tmp) <= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (strcmp(reply->str, tmp) >= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (strcmp(reply->str, tmp) != 0) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_BITMAP:
            i_value = atoi(reply->str);
            LM_DBG("comparing BITMAP %d %s %d\n", i_value, op, VAL_BITMAP(v));
            if (!strcmp(op, OP_EQ)) {
                if (i_value == VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_LT)) {
                if (i_value < VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_GT)) {
                if (i_value > VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_LEQ)) {
                if (i_value <= VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_GEQ)) {
                if (i_value >= VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_NEQ)) {
                if (i_value != VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_BITWISE_AND)) {
                if (i_value & VAL_BITMAP(v))
                    return 0;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_BLOB:
        default:
            LM_ERR("Unsupported val type %d\n", vtype);
            return -1;
    }
}

static int db_redis_convert_row(km_redis_con_t *con, db1_res_t* _r, const db_key_t *_k,
        const db_val_t *_v, const db_op_t *_op,
        redisReply *reply, const str *table_name, const db_key_t* _c, int _nc,
        int *manual_keys, int manual_keys_count) {
    db_val_t* dval;
    db_row_t* drow;
    size_t col;

    if (reply->type != REDIS_REPLY_ARRAY) {
        LM_ERR("Unexpected redis reply type, expecting array\n");
        return -1;
    }

    if (!reply->elements) {
        LM_DBG("skip empty row");
        return 0;
    }

    // manually filter non-matching replies
    for (col = 0; col < reply->elements; ++col) {
        if (col < manual_keys_count) {
            int idx = manual_keys[col];
            db_key_t k = _k[idx];
            db_val_t v = _v[idx];
            db_op_t o = _op[idx];
            LM_DBG("manually filtering key '%.*s'\n",
                    k->len, k->s);
            if (db_redis_compare_column(k, &v, o, reply->element[col]) != 0) {
                LM_DBG("column %lu does not match, ignore row\n", col);
                return 0;
            }
        }
    }

    RES_NUM_ROWS(_r) = RES_ROW_N(_r) = RES_NUM_ROWS(_r) + 1;
    drow = &(RES_ROWS(_r)[RES_NUM_ROWS(_r)-1]);

    if (db_allocate_row(_r, drow) != 0) {
        LM_ERR("Failed to allocate row %d\n", RES_NUM_ROWS(_r));
        return -1;
    }

    if (reply->elements - manual_keys_count > RES_COL_N(_r)) {
        LM_ERR("Invalid number of columns at row %d/%d, expecting %d, got %lu\n",
                RES_NUM_ROWS(_r), RES_ROW_N(_r), RES_COL_N(_r), reply->elements - manual_keys_count);
        return -1;
    }
    for (col = manual_keys_count; col < reply->elements; ++col) {
        size_t colidx = col - manual_keys_count;
        size_t redisidx = col;
        int coltype;
        redisReply *col_val = reply->element[redisidx];
        str *col_name = _c[colidx];

        LM_DBG("converting column #%lu of row #%d", colidx, RES_ROW_N(_r));

        if (col_val->type != REDIS_REPLY_STRING &&
            col_val->type != REDIS_REPLY_NIL) {

            LM_ERR("Invalid column value type in column '%.*s' of row %d, expecting string or null\n",
                    col_name->len, col_name->s, RES_NUM_ROWS(_r));
            return -1;
        }

        if (RES_NUM_ROWS(_r) == 1) {
            coltype = db_redis_schema_get_column_type(con, table_name, col_name);
            RES_TYPES(_r)[colidx] = coltype;
        } else {
            coltype = RES_TYPES(_r)[colidx];
        }

        dval = &(ROW_VALUES(drow)[colidx]);
        VAL_TYPE(dval) = coltype;

        if (col_val->type == REDIS_REPLY_NIL) {
            VAL_NULL(dval) = 1;
        } else {
            if (db_str2val(coltype, dval, col_val->str, strlen(col_val->str), 1) != 0) {
                LM_ERR("Failed to convert redis column '%.*s' to db value\n",
                        col_name->len, col_name->s);
                return -1;
            }
        }
    }

    return 0;
}

static int db_redis_perform_query(const db1_con_t* _h, km_redis_con_t *con, const db_key_t* _k,
        const db_val_t* _v, const db_op_t *_op, const db_key_t* _c,
        const int _n, const int _nc, db1_res_t** _r,
        redis_key_t **keys, int *keys_count,
        int **manual_keys, int *manual_keys_count, int do_table_scan) {

    redisReply *reply = NULL;
    redis_key_t *query_v = NULL;
    int num_rows = 0;
    redis_key_t *key;
    int j;

    *_r = db_redis_new_result();
    if (!*_r) {
        LM_ERR("Failed to allocate memory for result");
        goto error;
    }

    if (db_allocate_columns(*_r, _nc) != 0) {
        LM_ERR("Failed to allocate memory for result columns");
        goto error;
    }
    RES_NUM_ROWS(*_r) = RES_ROW_N(*_r) = 0;
    RES_COL_N(*_r) = _nc;

    if (!(*keys_count) && do_table_scan) {
        LM_DBG("performing full table scan\n");
        if (db_redis_scan_query_keys(con, CON_TABLE(_h), _k, _n,
                    keys, keys_count,
                    manual_keys, manual_keys_count) != 0) {
            LM_ERR("failed to scan query keys\n");
            goto error;
        }
    }

    for (key = *keys; key; key = key->next) {
        redis_key_t *tmp = NULL;
        str *keyname = &(key->key);

        num_rows++;

        LM_DBG("checking key '%s' in redis\n", keyname->s);

        if (db_redis_key_add_string(&query_v, "EXISTS", 6) != 0) {
            LM_ERR("Failed to add exists query to list\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, keyname) != 0) {
            LM_ERR("Failed to add key name to list\n");
            goto error;
        }
        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }
        tmp = db_redis_key_unshift(&query_v);
        if (tmp)
            db_redis_key_free(&tmp);

        // construct HMGET query
        if (_nc + (*manual_keys_count) == 0) {
            if (db_redis_key_prepend_string(&query_v, "HGETALL", 7) != 0) {
                LM_ERR("Failed to add hgetall query to list\n");
                goto error;
            }
        } else {
            if (db_redis_key_prepend_string(&query_v, "HMGET", 5) != 0) {
                LM_ERR("Failed to add hmget query to list\n");
                goto error;
            }
        }

        // we put the manual comparison columns first, so we can skip them
        // easily in result, for the cost of potential duplicate column returns
        for (j = 0; j < *manual_keys_count; ++j) {
            int idx = (*manual_keys)[j];
            str *k_name = _k[idx];
            if (db_redis_key_add_str(&query_v, k_name) != 0) {
                LM_ERR("Failed to add manual key to query list\n");
                goto error;
            }
        }
        for (j = 0; j < _nc; ++j) {
            str *k_name = _c[j];
            if (db_redis_key_add_str(&query_v, k_name) != 0) {
                LM_ERR("Failed to add manual key to query list\n");
                goto error;
            }
        }

        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }

        db_redis_key_free(&query_v);
        query_v = NULL;
    }

    // we allocate best case scenario (all rows match)
    RES_NUM_ROWS(*_r) = RES_ROW_N(*_r) = num_rows;
    if (db_allocate_rows(*_r) != 0) {
        LM_ERR("Failed to allocate memory for rows\n");
        return -1;
    }
    RES_COL_N(*_r) = _nc;
    // reset and increment in convert_row
    RES_NUM_ROWS(*_r) = RES_ROW_N(*_r) = 0;

    for (key = *keys; key; key = key->next) {
        // get reply for EXISTS query
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    con->con->errstr);
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        if (reply->integer == 0) {
            LM_DBG("key does not exist, returning no row for query\n");
            db_redis_free_reply(&reply);
            // also free next reply, as this is a null row for the HMGET
            db_redis_get_reply(con, (void**)&reply);
            db_redis_check_reply(con, reply, error);
            db_redis_free_reply(&reply);
            continue;
        }
        db_redis_free_reply(&reply);

        // get reply for actual HMGET query
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    con->con->errstr);
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        if (reply->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Unexpected reply, expected array\n");
            goto error;
        }
        LM_DBG("dumping full query reply for row\n");
        db_redis_dump_reply(reply);

        if (db_redis_convert_row(con, *_r, _k, _v, _op, reply, CON_TABLE(_h), _c, _nc, *manual_keys, *manual_keys_count)) {
            LM_ERR("Failed to convert redis reply for row\n");
            goto error;
        }
        db_redis_free_reply(&reply);
    }

    return 0;

error:
    LM_ERR("failed to perform the query\n");
    db_redis_key_free(&query_v);
    if(reply)
        db_redis_free_reply(&reply);
    if(_r && *_r) {
        db_redis_free_result((db1_con_t*)_h, *_r); *_r = NULL;
    }
    return -1;
}

static int db_redis_perform_delete(const db1_con_t* _h, km_redis_con_t *con, const db_key_t* _k,
        const db_val_t* _v, const db_op_t *_op, const int _n,
        redis_key_t *keys, int keys_count,
        int *manual_keys, int manual_keys_count, int do_table_scan) {

    int j = 0;
    redis_key_t *k = NULL;
    int type_keys_count = 0;
    int all_type_keys_count = 0;
    size_t col;

    redisReply *reply = NULL;
    redis_key_t *query_v = NULL;
    redis_key_t *type_keys = NULL;
    redis_key_t *all_type_keys = NULL;
    db_val_t *db_vals = NULL;
    db_key_t *db_keys = NULL;
    redis_key_t *type_key;

    if (!keys_count && do_table_scan) {
        LM_DBG("performing full table scan\n");
        if (db_redis_scan_query_keys(con, CON_TABLE(_h), _k, _n,
                    &keys, &keys_count,
                    &manual_keys, &manual_keys_count) != 0) {
            LM_ERR("failed to scan query keys\n");
            goto error;
        }
    }

    // TODO: this should be moved to redis_connection structure
    // and be parsed at startup:
    //
    // fetch list of keys in all types
    if (db_redis_get_keys_for_all_types(con, CON_TABLE(_h),
                &all_type_keys, &all_type_keys_count) != 0) {
            LM_ERR("failed to get full list of type keys\n");
            goto error;
    }

    LM_DBG("+++ delete all keys\n");
    for (k = keys; k; k = k->next) {
        redis_key_t *all_type_key;
        str *key = &k->key;
        redis_key_t *tmp = NULL;
        int row_match;
        LM_DBG("+++ delete key '%.*s'\n", key->len, key->s);

        if (db_redis_key_add_string(&query_v, "EXISTS", 6) != 0) {
            LM_ERR("Failed to add exists command to pre-delete query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, key) != 0) {
            LM_ERR("Failed to add key name to pre-delete query\n");
            goto error;
        }

        // TODO: pipeline commands!
        reply = db_redis_command_argv(con, query_v);
        db_redis_check_reply(con, reply, error);
        if (reply->integer == 0) {
            LM_DBG("key does not exist in redis, skip deleting\n");
            db_redis_free_reply(&reply);
            continue;
        }
        db_redis_free_reply(&reply);
        tmp = db_redis_key_unshift(&query_v);
        if (tmp)
            db_redis_key_free(&tmp);

        if (db_redis_key_prepend_string(&query_v, "HMGET", 5) != 0) {
            LM_ERR("Failed to set hmget command to pre-delete query\n");
            goto error;
        }

        // add all manual keys to query
        for (j = 0; j < manual_keys_count; ++j) {
            int idx = manual_keys[j];
            str *col = _k[idx];

            if (db_redis_key_add_str(&query_v, col) != 0) {
                LM_ERR("Failed to add manual key to pre-delete query\n");
                goto error;
            }
        }
        // add all type keys to query
        for (all_type_key = all_type_keys; all_type_key; all_type_key = all_type_key->next) {
            if (db_redis_key_add_str(&query_v, &all_type_key->key) != 0) {
                LM_ERR("Failed to add type key to pre-delete query\n");
                goto error;
            }
        }
        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);

        LM_DBG("dumping full query reply\n");
        db_redis_dump_reply(reply);

        // manually filter non-matching replies
        row_match = 1;
        for (col = 0; col < reply->elements; ++col) {
            if (col < manual_keys_count) {
                int idx = manual_keys[col];
                db_key_t k = _k[idx];
                db_val_t v = _v[idx];
                db_op_t o = _op[idx];
                LM_DBG("manually filtering key '%.*s'\n",
                        k->len, k->s);
                if (db_redis_compare_column(k, &v, o, reply->element[col]) != 0) {
                    LM_DBG("column %lu does not match, ignore row\n", col);
                    row_match = 0;
                    break;
                }
            }
        }
        if (!row_match) {
            db_redis_free_reply(&reply);
            continue;
        } else {
            LM_DBG("row matches manual filtering, proceed with deletion\n");
        }

        db_keys = (db_key_t*) pkg_malloc(all_type_keys_count * sizeof(db_key_t));
        if (!db_keys) {
            LM_ERR("Failed to allocate memory for db type keys\n");
            goto error;
        }
        for (j = 0, tmp = all_type_keys; tmp; ++j, tmp = tmp->next) {
            db_keys[j] = &tmp->key;
        }

        db_vals = (db_val_t*) pkg_malloc(all_type_keys_count * sizeof(db_val_t));
        if (!db_vals) {
            LM_ERR("Failed to allocate memory for manual db vals\n");
            goto error;
        }

        for (j = 0, all_type_key = all_type_keys; all_type_key; ++j, all_type_key = all_type_key->next) {
            db_val_t *v = &(db_vals[j]);
            str *key = &all_type_key->key;
            char *value = reply->element[manual_keys_count + j]->str;
            int coltype = db_redis_schema_get_column_type(con, CON_TABLE(_h), key);
            if (value == NULL) {
                VAL_NULL(v) = 1;
            } else if (db_str2val(coltype, v, value, strlen(value), 0) != 0) {
                LM_ERR("Failed to convert redis reply column to db value\n");
                goto error;
            }
        }
        if (db_redis_build_type_keys(con, CON_TABLE(_h), db_keys, db_vals, all_type_keys_count,
                    &type_keys, &type_keys_count) != 0) {
            LM_ERR("failed to build type keys\n");
            goto error;
        }
        pkg_free(db_keys);
        db_keys = NULL;
        pkg_free(db_vals);
        db_vals = NULL;
        db_redis_free_reply(&reply);

        if (db_redis_key_add_string(&query_v, "DEL", 3) != 0) {
            LM_ERR("Failed to add del command to delete query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, key) != 0) {
            LM_ERR("Failed to add key to delete query\n");
            goto error;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);

        for (type_key = type_keys; type_key; type_key = type_key->next) {
            if (db_redis_key_add_string(&query_v, "SREM", 4) != 0) {
                LM_ERR("Failed to add srem command to post-delete query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, &type_key->key) != 0) {
                LM_ERR("Failed to add key to delete query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, key) != 0) {
                LM_ERR("Failed to add key to delete query\n");
                goto error;
            }
            reply = db_redis_command_argv(con, query_v);
            db_redis_key_free(&query_v);
            db_redis_check_reply(con, reply, error);
            db_redis_free_reply(&reply);
        }

        //db_redis_key_free(&type_keys);
        LM_DBG("+++ done with loop '%.*s'\n", k->key.len, k->key.s);
    }
    db_redis_key_free(&type_keys);
    db_redis_key_free(&all_type_keys);
    db_redis_key_free(&query_v);

    return 0;

error:
    LM_ERR("failed to perform the delete\n");
    if(reply)
        db_redis_free_reply(&reply);
    if (db_keys)
        pkg_free(db_keys);
    if (db_vals)
        pkg_free(db_vals);
    db_redis_key_free(&query_v);
    db_redis_key_free(&type_keys);
    db_redis_key_free(&all_type_keys);
    return -1;
}

static int db_redis_perform_update(const db1_con_t* _h, km_redis_con_t *con, const db_key_t* _k,
        const db_val_t* _v, const db_op_t *_op, const db_key_t* _uk, const db_val_t *_uv,
        const int _n, const int _nu,
        redis_key_t **keys, int *keys_count,
        int **manual_keys, int *manual_keys_count, int do_table_scan) {

    redisReply *reply = NULL;
    redis_key_t *query_v = NULL;
    int update_queries = 0;
    redis_key_t *key;
    int i;
    int j;
    size_t col;

    if (!(*keys_count) && do_table_scan) {
        LM_DBG("performing full table scan\n");
        if (db_redis_scan_query_keys(con, CON_TABLE(_h), _k, _n,
                    keys, keys_count,
                    manual_keys, manual_keys_count) != 0) {
            LM_ERR("failed to scan query keys\n");
            goto error;
        }
    }

    for (key = *keys; key; key = key->next) {
        str *keyname = &key->key;

        LM_DBG("fetching row for '%.*s' from redis\n", keyname->len, keyname->s);


        if (db_redis_key_add_string(&query_v, "EXISTS", 6) != 0) {
            LM_ERR("Failed to set exists command to pre-update exists query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, keyname) != 0) {
            LM_ERR("Failed to add key name to pre-update exists query\n");
            goto error;
        }
        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }
        db_redis_key_free(&query_v);

        // construct HMGET query
        if ((*manual_keys_count) == 0) {
            if (db_redis_key_add_string(&query_v, "HGETALL", 7) != 0) {
                LM_ERR("Failed to set hgetall command to pre-update hget query\n");
                goto error;
            }
            // TODO: actually we wouldn't have to fetch it at all, but then we'd
            // have to mark this key telling to not fetch reply of HMGET after
            // EXISTS returns false!
        } else {
            if (db_redis_key_add_string(&query_v, "HMGET", 5) != 0) {
                LM_ERR("Failed to set hgetall command to pre-update hget query\n");
                goto error;
            }
        }
        if (db_redis_key_add_str(&query_v, keyname) != 0) {
            LM_ERR("Failed to add key name to pre-update exists query\n");
            goto error;
        }

        for (j = 0; j < *manual_keys_count; ++j) {
            int idx = (*manual_keys)[j];
            str *k_name = _k[idx];
            if (db_redis_key_add_str(&query_v, k_name) != 0) {
                LM_ERR("Failed to add manual key name to hget query\n");
                goto error;
            }
        }

        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }
        db_redis_key_free(&query_v);
    }

/*
    key_value = (str*)pkg_malloc(_nu * sizeof(str));
    if (!key_value) {
        LM_ERR("Failed to allocate memory for key buffer\n");
        goto error;
    }
    memset(key_value, 0, _nu * sizeof(str));

    col_value = (str*)pkg_malloc(_nu * sizeof(str));
    if (!col_value) {
        LM_ERR("Failed to allocate memory for column buffer\n");
        goto error;
    }
    memset(col_value, 0, _nu * sizeof(str));
    */


    for (key = *keys; key; key = key->next) {
        int row_match;

        LM_DBG("fetching replies for '%.*s' from redis\n", key->key.len, key->key.s);

        // get reply for EXISTS query
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    con->con->errstr);
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        if (reply->integer == 0) {
            LM_DBG("key does not exist, returning no row for query\n");
            db_redis_free_reply(&reply);
            // also free next reply, as this is a null row for the HMGET
            LM_DBG("also fetch hmget reply after non-existent key\n");
            db_redis_get_reply(con, (void**)&reply);
            db_redis_check_reply(con, reply, error);
            db_redis_free_reply(&reply);
            LM_DBG("continue fetch reply loop\n");
            continue;
        }
        db_redis_free_reply(&reply);

        // get reply for actual HMGET query
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    con->con->errstr);
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        if (reply->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Unexpected reply, expected array\n");
            goto error;
        }
        LM_DBG("dumping full query reply for row\n");
        db_redis_dump_reply(reply);

        // manually filter non-matching replies
        row_match = 1;
        for (col = 0; col < reply->elements; ++col) {
            if (col < *manual_keys_count) {
                int idx = (*manual_keys)[col];
                db_key_t k = _k[idx];
                db_val_t v = _v[idx];
                db_op_t o = _op[idx];
                LM_DBG("manually filtering key '%.*s'\n",
                        k->len, k->s);
                if (db_redis_compare_column(k, &v, o, reply->element[col]) != 0) {
                    LM_DBG("column %lu does not match, ignore row\n", col);
                    row_match = 0;
                    break;
                }
            }
        }
        db_redis_free_reply(&reply);
        if (!row_match) {
            continue;
        } else {
            LM_DBG("row matches manual filtering, proceed with update\n");
        }

        if (db_redis_key_add_string(&query_v, "HMSET", 5) != 0) {
            LM_ERR("Failed to add hmset command to update query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &key->key) != 0) {
            LM_ERR("Failed to add key to update query\n");
            goto error;
        }

        for (i = 0; i < _nu; ++i) {
            str *k = _uk[i];
            str v = {NULL, 0};

            if (db_redis_val2str(&(_uv[i]), &v) != 0) {
                LM_ERR("Failed to convert update value to string\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, k) != 0) {
                LM_ERR("Failed to add key to update query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, &v) != 0) {
                LM_ERR("Failed to add key to update query\n");
                goto error;
            }
            if (v.s)
                pkg_free(v.s);
        }
        update_queries++;
        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }

        db_redis_key_free(&query_v);
    }

    LM_DBG("getting replies for %d queries\n", update_queries);

    for (i = 0; i < update_queries; ++i) {
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    con->con->errstr);
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);
    }

    LM_DBG("done performing update\n");

    return 0;

error:
    LM_ERR("failed to perform the update\n");
    if (reply)
        db_redis_free_reply(&reply);
    db_redis_key_free(&query_v);
    return -1;
}


/*
 * Query table for specified rows
 * _h: structure representing database connection
 * _k: key names
 * _op: operators
 * _v: values of the keys that must match
 * _c: column names to return
 * _n: number of key=values pairs to compare
 * _nc: number of columns to return
 * _o: order by the specified column
 */
int db_redis_query(const db1_con_t* _h, const db_key_t* _k, const db_op_t* _op,
        const db_val_t* _v, const db_key_t* _c, const int _n, const int _nc,
        const db_key_t _o, db1_res_t** _r)
{
    km_redis_con_t *con = NULL;
    int free_op = 0;
    int do_table_scan = 0;

    redis_key_t *keys = NULL;
    int keys_count = 0;
    int *manual_keys = NULL;
    int manual_keys_count = 0;
    db_op_t *query_ops = NULL;
    int i;

    // TODO: implement order-by
    // TODO: optimize mapping-based manual post-check (remove check for keys already
    // in type query key)

    if (!_r) {
        LM_ERR("db result is null\n");
        return -1;
    }
    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("querying prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    *_r = NULL;

    // check if we have a version query, and return version directly from
    // schema instead of loading it from redis
    if (_nc == 1 && _n >= 1 && VAL_TYPE(&_v[0]) == DB1_STR &&
            CON_TABLE(_h)->len == strlen("version") && strncmp(CON_TABLE(_h)->s, "version", CON_TABLE(_h)->len) == 0 &&
            _k[0]->len == strlen("table_name") && strncmp(_k[0]->s, "table_name", _k[0]->len) == 0 &&
            _c[0]->len == strlen("table_version") && strncmp(_c[0]->s, "table_version", _c[0]->len) == 0) {

        if (db_redis_return_version(_h, con, &VAL_STR(&_v[0]), _r) == 0) {
            return 0;
        }
        // if we fail to return a version from the schema, go query the table, just in case
    }

    free_op = 0;

    if (_op == NULL) {
        char *op = "=";
        free_op = 1;
        query_ops = (db_op_t*)pkg_malloc(_n * sizeof(db_op_t));
        if (!query_ops) {
            LM_ERR("Failed to allocate memory for query op list\n");
            goto error;
        }
        for (i = 0; i < _n; ++i) {
            query_ops[i] = op;
        }
    } else {
        query_ops = (db_op_t*)_op;
    }

    if (_n > 0) {
        if (db_redis_build_query_keys(con, CON_TABLE(_h), _k, _v, query_ops, _n,
                    &keys, &keys_count, &manual_keys, &manual_keys_count, &do_table_scan) != 0) {
            LM_ERR("failed to build query keys\n");
            goto error;
        }
        if (!keys_count) {
            if (do_table_scan) {
                LM_DBG("unable to build query keys, falling back to full table scan\n");
            } else {
                LM_DBG("query keys not member of suitable mapping, skip full table scan\n");
            }
        }
    } else {
        LM_DBG("no columns given to build query keys, falling back to full table scan\n");
        keys_count = 0;
    }

    if (db_redis_perform_query(_h, con, _k, _v, query_ops, _c, _n, _nc, _r,
        &keys, &keys_count, &manual_keys, &manual_keys_count, do_table_scan) != 0) {
        goto error;
    }

    LM_DBG("done performing query\n");

    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);

    if (manual_keys) {
        pkg_free(manual_keys);
    }

    db_redis_consume_replies(con);
    return 0;

error:
    LM_ERR("failed to do the query\n");
    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys) {
        pkg_free(manual_keys);
    }
    db_redis_consume_replies(con);


    return -1;
}

/*
 * Execute a raw SQL query
 */
int db_redis_raw_query(const db1_con_t* _h, const str* _s, db1_res_t** _r)
{
    LM_DBG("perform redis raw query\n");
    return -1;
}

/*
 * Insert a row into specified table
 * _h: structure representing database connection
 * _k: key names
 * _v: values of the keys
 * _n: number of key=value pairs
 */
int db_redis_insert(const db1_con_t* _h, const db_key_t* _k, const db_val_t* _v, const int _n)
{
    km_redis_con_t *con = NULL;
    redis_key_t *key = NULL;
    int keys_count = 0;
    redis_key_t *type_keys = NULL;
    int type_keys_count = 0;
    redis_key_t *query_v = NULL;
    redisReply *reply = NULL;
    int i;
    redis_key_t *k;

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("inserting to prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    if (db_redis_build_entry_keys(con, CON_TABLE(_h), _k, _v, _n,
                &key, &keys_count) != 0) {
        LM_ERR("failed to build entry keys\n");
        goto error;
    }
    if (db_redis_build_type_keys(con, CON_TABLE(_h), _k, _v, _n,
                &type_keys, &type_keys_count) != 0) {
        LM_ERR("failed to build type keys\n");
        goto error;
    }

    if (db_redis_key_add_string(&query_v, "HMSET", 5) != 0) {
        LM_ERR("Failed to add hmset command to insert query\n");
        goto error;
    }
    if (db_redis_key_add_str(&query_v, &key->key) != 0) {
        LM_ERR("Failed to add key to insert query\n");
        goto error;
    }

    for (i = 0; i < _n; ++i) {
        str *k = _k[i];
        str v;

        if (db_redis_key_add_str(&query_v, k) != 0) {
            LM_ERR("Failed to add column name to insert query\n");
            goto error;
        }
        if (db_redis_val2str(&(_v[i]), &v) != 0) {
            LM_ERR("Failed to allocate memory for query value\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &v) != 0) {
            LM_ERR("Failed to add column value to insert query\n");
            goto error;
        }
        if (v.s)
            pkg_free(v.s);
    }

    reply = db_redis_command_argv(con, query_v);
    db_redis_key_free(&query_v);
    db_redis_check_reply(con, reply, error);
    db_redis_free_reply(&reply);

    for (k = type_keys; k; k = k->next) {
        str *type_key = &k->key;

        LM_DBG("inserting entry key '%.*s' to type map '%.*s'\n",
            key->key.len, key->key.s, type_key->len, type_key->s);

        if (db_redis_key_add_string(&query_v, "SADD", 4) != 0) {
            LM_ERR("Failed to set sadd command to post-insert query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, type_key) != 0) {
            LM_ERR("Failed to add map key to post-insert query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &key->key) != 0) {
            LM_ERR("Failed to set entry key to post-insert query\n");
            goto error;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);
    }

    db_redis_key_free(&key);
    db_redis_key_free(&type_keys);
    db_redis_consume_replies(con);

    return 0;

error:
    db_redis_key_free(&key);
    db_redis_key_free(&type_keys);
    db_redis_key_free(&query_v);

    if (reply)
        db_redis_free_reply(&reply);

    db_redis_consume_replies(con);

    LM_ERR("failed to do the insert\n");
    return -1;
}

/*
 * Delete a row from the specified table
 * _h: structure representing database connection
 * _k: key names
 * _o: operators
 * _v: values of the keys that must match
 * _n: number of key=value pairs
 */
int db_redis_delete(const db1_con_t* _h, const db_key_t* _k,
        const db_op_t* _op, const db_val_t* _v, const int _n)
{
    km_redis_con_t *con = NULL;
    redis_key_t *keys = NULL;
    int keys_count = 0;
    int *manual_keys = NULL;
    int manual_keys_count = 0;
    int free_op = 0;
    int do_table_scan = 0;
    db_op_t *query_ops = NULL;
    int i;

    // TODO: optimize mapping-based manual post-check (remove check for keys already
    // in type query key)

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("deleting from prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    free_op = 0;

    if (_op == NULL) {
        char *op = "=";
        free_op = 1;
        query_ops = (db_op_t*)pkg_malloc(_n * sizeof(db_op_t));
        if (!query_ops) {
            LM_ERR("Failed to allocate memory for query op list\n");
            goto error;
        }
        for (i = 0; i < _n; ++i) {
            query_ops[i] = op;
        }
    } else {
        query_ops = (db_op_t*)_op;
    }

    if (_n > 0) {
        if (db_redis_build_query_keys(con, CON_TABLE(_h), _k, _v, query_ops, _n,
                    &keys, &keys_count, &manual_keys, &manual_keys_count, &do_table_scan) != 0) {
            LM_ERR("failed to build query keys\n");
            goto error;
        }
        if (!keys_count) {
            if (do_table_scan) {
                LM_DBG("unable to build query keys, falling back to full table scan\n");
            } else {
                LM_DBG("query keys not member of suitable mapping, skip full table scan\n");
            }
        }
    } else {
        LM_DBG("no columns given to build query keys, falling back to full table scan\n");
        keys_count = 0;
    }

    if (db_redis_perform_delete(_h, con, _k, _v, query_ops, _n,
        keys, keys_count, manual_keys, manual_keys_count, do_table_scan) != 0) {
        goto error;
    }

    LM_DBG("done performing delete\n");

    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys)
        pkg_free(manual_keys);
    db_redis_consume_replies(con);

    return 0;

error:
    LM_ERR("failed to do the query\n");
    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys)
        pkg_free(manual_keys);
    db_redis_consume_replies(con);
    return -1;
}

/*
 * Update some rows in the specified table
 * _h: structure representing database connection
 * _k: key names
 * _op: operators
 * _v: values of the keys that must match
 * _uk: updated columns
 * _uv: updated values of the columns
 * _n: number of key=value pairs
 * _un: number of columns to update
 */
int db_redis_update(const db1_con_t* _h, const db_key_t* _k,
        const db_op_t* _op, const db_val_t* _v, const db_key_t* _uk,
        const db_val_t* _uv, const int _n, const int _nu)
{
    km_redis_con_t *con = NULL;
    int free_op = 0;
    int do_table_scan = 0;

    redis_key_t *keys = NULL;
    int keys_count = 0;
    int *manual_keys = NULL;
    int manual_keys_count = 0;
    db_op_t *query_ops = NULL;
    int i;

    // TODO: optimize mapping-based manual post-check (remove check for keys already
    // in type query key)

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("updating prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    free_op = 0;

    if (_op == NULL) {
        char *op = "=";
        free_op = 1;
        query_ops = (db_op_t*)pkg_malloc(_n * sizeof(db_op_t));
        if (!query_ops) {
            LM_ERR("Failed to allocate memory for query op list\n");
            goto error;
        }
        for (i = 0; i < _n; ++i) {
            query_ops[i] = op;
        }
    } else {
        query_ops = (db_op_t*)_op;
    }

    if (_n > 0) {
        if (db_redis_build_query_keys(con, CON_TABLE(_h), _k, _v, query_ops, _n,
                    &keys, &keys_count, &manual_keys, &manual_keys_count, &do_table_scan) != 0) {
            LM_ERR("failed to build query keys\n");
            goto error;
        }
        if (!keys_count) {
            if (do_table_scan) {
                LM_DBG("unable to build query keys, falling back to full table scan\n");
            } else {
                LM_DBG("query keys not member of suitable mapping, skip full table scan\n");
            }
        }
    } else {
        LM_DBG("no columns given to build query keys, falling back to full table scan\n");
        keys_count = 0;
    }

    if (db_redis_perform_update(_h, con, _k, _v, query_ops, _uk, _uv, _n, _nu,
        &keys, &keys_count, &manual_keys, &manual_keys_count, do_table_scan) != 0) {
        goto error;
    }

    LM_DBG("done performing update\n");

    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);

    if (manual_keys) {
        pkg_free(manual_keys);
    }
    db_redis_consume_replies(con);
    return 0;

error:
    LM_ERR("failed to do the query\n");
    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys) {
        pkg_free(manual_keys);
    }
    db_redis_consume_replies(con);
    return -1;
}

/*
 * Just like insert, but replace the row if it exists
 */
int db_redis_replace(const db1_con_t* _h, const db_key_t* _k,
        const db_val_t* _v, const int _n, const int _un, const int _m) {
    LM_DBG("perform redis replace\n");
    return -1;
}

/*
 * Store name of table that will be used by
 * subsequent database functions
 */
int db_redis_use_table(db1_con_t* _h, const str* _t) {
    return db_use_table(_h, _t);
}

int db_redis_free_result(db1_con_t* _h, db1_res_t* _r) {
    LM_DBG("perform redis free result\n");

    if(!_r)
        return -1;
    db_free_result(_r);
    return 0;
}
